package com.letv.dataprocess.cp_info;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TMemoryBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import serving.ContentProviderAttribute;
import serving.Thumbnail;
import serving.VideoIndex;

import com.letv.bean.CpInfoBean;
import com.letv.bean.VideoIndexBean;
import com.letv.common.ConnectionManager;
import com.letv.conf.ConfigFactory;
import com.letv.dataprocess.DataProcessResult;
import com.letv.mms.cache.factory.CacheFactory;
import com.letv.util.DateUtil;
import com.letv.util.SQLUtil;

/**
 * process and put cp_info into cbase
 */
public class CpInfoDataProcessTask {

    /**
     * int, if add new album, cms will send this action to offline
     */
    public static final int ADD_ALBUM_ACTION = 8881001;

    /**
     * int, if delete album, cms will send this action to offline
     */
    public static final int DELETE_ALBUM_ACTION = 8881003;

    /**
     * cp mq if open.
     * 0 : not open
     * 1 : open
     */
    public static final String CP_MQ_IF_OPEN = ConfigFactory.getString("mqcp.isopen");

    /**
     * int, max video count of one cp request
     */
    private static final int MAX_VIDEO = 500;

    /**
     * Logger
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(CpInfoDataProcessTask.class);

    /**
     * process data into cbase for batch
     * @param listIndex int, index in the worker list.It's sign of this worker
     * @param beginId int, not include this id when select from data base
     * @param endId int, include this id when select from data base
     * @return DataProcessResult, this worker's data process infomation.return null while has errors
     */
    public DataProcessResult processIntoCbaseBatch(int listIndex, int beginId, int endId) {
        if(beginId >= endId) {
            return null;
        }

        ConnectionManager connectionManager = null;
        try {
            connectionManager = ConnectionManager.getInstance();
        } catch (Exception e) {
            LOGGER.error("worker " + listIndex + " cp_info fail to process date into cbase,connection get instance error!", e);
            return null;
        }

        Connection connectionFromLeso = null;
        try {
            connectionFromLeso = connectionManager.getConnectionFromLeso();
        } catch (Exception e) {
            LOGGER.error("worker " + listIndex + " cp_info fail to process date into cbase,connection get connectionFromLeso error!", e);
            return null;
        }

        List<CpInfoBean> cpInfoList = null;
        try {
            cpInfoList = SQLUtil.getCpInfoRange(connectionFromLeso, beginId, endId);
        } catch(Exception e) {
            LOGGER.error("worker " + listIndex + " has errors while select cp_info data from database", e);
        } finally{
            if(!SQLUtil.close(connectionFromLeso)) {
                LOGGER.error("fail to close connectionFromLeso when get cp_info range");
            }
        }

        if(null == cpInfoList || cpInfoList.size() == 0) {
            LOGGER.error("worker " + listIndex + "select none cp_info record from database");
            return null;
        }
        LOGGER.info("worker " + listIndex + "select " + cpInfoList.size() + " cp_info record from database");
        DataProcessResult result = new DataProcessResult();
        result.setBegin(beginId);
        result.setEnd(endId);
        result.setRecordsCnt(cpInfoList.size());

        int successIntoCbaseCount = 0;
        int failIntoCbaseCount = 0;

        Connection connectionFromLocal = null;
        try {
            connectionFromLocal = connectionManager.getConnection();
            for(CpInfoBean cpInfoBean : cpInfoList) {
                String albumIdList = this.getAlbumIdListByCpOId(cpInfoBean.getLetv_original_id(), null, connectionFromLocal);
                cpInfoBean.setAlbum_id_list(albumIdList);
                cpInfoBean.setVideo_index_list(this.getVideoIndexListByCpOId(cpInfoBean.getLetv_original_id(), connectionFromLocal));
            }
        } catch(Exception e) {
            LOGGER.error("worker " + listIndex + " has errors while get cp_album_id_list from database", e);
        } finally{
            if(!SQLUtil.close(connectionFromLocal)) {
                LOGGER.error("fail to close connectionFromLocal when get cp_album_id_list");
            }
        }

        for(CpInfoBean cpInfoBean : cpInfoList) {
            ContentProviderAttribute cPInfo = this.transformToThriftObject(cpInfoBean);

            byte[] thriftSerializeResult = this.thriftSerialize(cPInfo);
            if(null == thriftSerializeResult) {
                failIntoCbaseCount++;
                continue;
            }

            try {
                CacheFactory.getCache().set(cpInfoBean.getLetv_original_id(), thriftSerializeResult, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
                LOGGER.info("thrift object cp info put into cbase successly,key=" + cpInfoBean.getLetv_original_id());
            } catch(Exception e) {
                LOGGER.error("cpId=" + cpInfoBean.getId() + " has errors to put in cbase", e);
                failIntoCbaseCount++;
                continue;
            }
            successIntoCbaseCount++;
        }
        result.setCbaseSuccess(successIntoCbaseCount);
        result.setCbasefail(failIntoCbaseCount);
        return result;
    }

    /**
     * process into cbase by albumId
     * @param albumId String
     * @param action int, album change mode.now have two changes:add or delete
     */
    public void processIntoCbaseByAlbumId(String albumId, int action) {
        if(StringUtils.isBlank(albumId)) {
            return;
        }
        LOGGER.info("begin to process cp info by albumId=" + albumId + ",action=" + action);

        ConnectionManager connectionManager = null;
        Connection connectionFromLocal = null;
        try {
            connectionManager = ConnectionManager.getInstance();
            connectionFromLocal = connectionManager.getConnection();
        } catch (Exception e) {
            LOGGER.error("get connection error! albumId = " + albumId, e);
            return;
        }

        Integer cpId = null;
        try {
            cpId = SQLUtil.getCpIdByAlbumId(connectionFromLocal, albumId);
        } catch(Exception e) {
            LOGGER.error("have errors while get cpid list by albumId from database", e);
            return;
        }  finally{
            if(!SQLUtil.close(connectionFromLocal)) {
                LOGGER.error("fail to close connectionFromLeso");
            }
        }
        if(null == cpId) {
            return;
        }
        this.processIntoCbaseByCpId(cpId.intValue(), albumId, action);
    }

    /**
     * process data into cbase by cp_info id
     * @param cpId int, cp info id
     * @param albumId String, the id of changed album
     * @param action int, album change mode.now have two changes:add or delete
     */
    public void processIntoCbaseByCpId(int cpId, String albumId, int action) {
        if(StringUtils.isBlank(albumId)) {
            return;
        }

        ConnectionManager connectionManager = null;
        Connection connectionFromLeso = null;
        Connection connectionFromLocal = null;
        try {
            connectionManager = ConnectionManager.getInstance();
            connectionFromLeso = connectionManager.getConnectionFromLeso();
            connectionFromLocal = connectionManager.getConnection();
        } catch (Exception e) {
            LOGGER.error("cp_info fail to process date into cbase by id,get connection error!cp_id=" + cpId, e);
            return;
        }

        List<CpInfoBean> cpInfoList = null;
        try {
            cpInfoList = SQLUtil.getCpInfoRange(connectionFromLeso, cpId - 1, cpId);
        } catch(Exception e) {
            LOGGER.error("have errors while select cp_info data from database by id.cp_id=" + cpId, e);
        } finally{
            if(!SQLUtil.close(connectionFromLeso)) {
                LOGGER.error("fail to close connectionFromLeso when get cp_info by id.cp_id=" + cpId);
            }
        }

        if(null == cpInfoList || cpInfoList.size() != 1 || null == cpInfoList.get(0)) {
            LOGGER.error("select none cp_info record or have errors from database by id.cp_id=" + cpId);
            return;
        }
        CpInfoBean cpInfoBean = cpInfoList.get(0);

        try {
            if(CpInfoDataProcessTask.DELETE_ALBUM_ACTION != action) {
                albumId = null;
            }
            String albumIdList = this.getAlbumIdListByCpOId(cpInfoBean.getLetv_original_id(), albumId, connectionFromLocal);
            cpInfoBean.setAlbum_id_list(albumIdList);
            cpInfoBean.setVideo_index_list(this.getVideoIndexListByCpOId(cpInfoBean.getLetv_original_id(), connectionFromLocal));
        } catch(Exception e) {
            LOGGER.error("have errors while get cp_album_id_list from database by id.cp_id=" + cpId, e);
        } finally{
            if(!SQLUtil.close(connectionFromLocal)) {
                LOGGER.error("fail to close connectionFromLocal when get cp_album_id_list by id");
            }
        }

        ContentProviderAttribute cPInfo = this.transformToThriftObject(cpInfoBean);
        byte[] thriftSerializeResult = this.thriftSerialize(cPInfo);
        if(null == thriftSerializeResult) {
            return;
        }

        try {
            CacheFactory.getCache().set(cpInfoBean.getLetv_original_id(), thriftSerializeResult, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
            LOGGER.info("thrift object cp info put into cbase by id successly,key=" + cpInfoBean.getLetv_original_id() + ", changeAction=" + action);
        } catch(Exception e) {
            LOGGER.error("cpId=" + cpInfoBean.getId() + " has errors to put in cbase", e);
        }
    }

    /**
     * get album id list by cp letv_original_id
     * @param cpLetvOriginalId String
     * @param deleteAlbumId String, this albumId is deleted by increased message actually;the albumIdList in cp_info will delete it.
     *                              if needn't delete albumId,this parameter will be null.
     * @param connectionFromLocal Connection, connection from local
     * @return String, the album_id list belong to this cp,separated by English comma;Example:1,2,3,4,5;return null if have errors or no record
     */
    private String getAlbumIdListByCpOId(String cpLetvOriginalId, String deleteAlbumId, Connection connectionFromLocal) {
        if(StringUtils.isBlank(cpLetvOriginalId) || null == connectionFromLocal) {
            return null;
        }
        List<String> albumIdList = null;
        try {
            albumIdList = SQLUtil.getAlbumIdListByCpOId(connectionFromLocal, cpLetvOriginalId);
        } catch(Exception e) {
            LOGGER.error("has errors while get album id list by cp id from database", e);
        }
        if(null == albumIdList || albumIdList.size() == 0) {
            return null;
        }

        if(StringUtils.isNotBlank(deleteAlbumId)) {
            try {
                for(int i = 0; i < albumIdList.size(); i++) {
                    if(albumIdList.get(i).equals(deleteAlbumId)) {
                        albumIdList.remove(i);
                        break;
                    }
                }
            } catch(Exception e) {
                LOGGER.error("error to delete albumId from cpInfo, albumId = " + deleteAlbumId + ". cp_info_id = " + cpLetvOriginalId, e);
            }
        }

        return StringUtils.join(albumIdList.toArray(), ",");
    }

    /**
     * 得到cp下单视频排好序的索引列表
     * @param cpLetvOriginalId String, cp全局id
     * @param connectionFromLocal Connection, 离线计算本地连接
     * @return List<VideoIndexBean>, 为空或出错则返回null
     */
    private List<VideoIndexBean> getVideoIndexListByCpOId(String cpLetvOriginalId, Connection connectionFromLocal) {
        if(StringUtils.isBlank(cpLetvOriginalId) || null == connectionFromLocal) {
            return null;
        }

        List<Map<String, Object>> svideos = null;
        try {
            svideos = SQLUtil.getSvideoByCpOId(connectionFromLocal, cpLetvOriginalId);
        } catch(Exception e) {
            LOGGER.error("error to get video index list by cp oid, cp_oid=" + cpLetvOriginalId, e);
        }
        if(null == svideos) {
            return null;
        }

        /**
         * 排序保证未完结专辑在完结专辑之前
         */
        List<Map<String, Object>> sortList = new ArrayList<Map<String, Object>>();
        List<Map<String, Object>> unEndList = new ArrayList<Map<String, Object>>();
        List<Map<String, Object>> isEndList = new ArrayList<Map<String, Object>>();
        for(int i = 0; i < svideos.size(); i++) {
            if(null == svideos.get(i) || null == svideos.get(i).get("id") || null == svideos.get(i).get("playurls") || null == svideos.get(i).get("category")) {
                continue;
            }
            try {
                String isEnd = "";
                if(null == svideos.get(i).get("isEnd")) {
                    isEnd = "1";
                } else {
                    isEnd = svideos.get(i).get("isEnd").toString();
                }
                if("0".equals(isEnd)) {
                    unEndList.add(svideos.get(i));
                } else {
                    isEndList.add(svideos.get(i));
                }
            } catch(Exception e) {
                isEndList.add(svideos.get(i));
            }
        }
        sortList.addAll(unEndList);
        sortList.addAll(isEndList);
        if(null == sortList || sortList.size() == 0) {
            return null;
        }

        List<List<VideoIndexBean>> albumList = new ArrayList<List<VideoIndexBean>>();
        int maxLength = 0;
        for(int i = 0; i < sortList.size(); i++) {
            try {
                List<VideoIndexBean> album = new ArrayList<VideoIndexBean>();
                long albumId = Long.parseLong(sortList.get(i).get("id").toString());
                String playurls = sortList.get(i).get("playurls").toString();
                String category = sortList.get(i).get("category").toString();
                boolean desc = category.equals("11") || category.equals("16") || category.equals("101") || category.equals("4");
                String[] playurlArray = playurls.split(";");
                if(playurlArray.length > maxLength) {
                    maxLength = playurlArray.length;
                }
                for(int j = 0; j < playurlArray.length; j++) {
                    if(desc) {
                        album.add(new VideoIndexBean(albumId, j));
                    } else {
                        album.add(new VideoIndexBean(albumId, playurlArray.length - 1 - j));
                    }
                }
                albumList.add(album);
            } catch(Exception e) {
                continue;
            }
        }
        if(null == albumList || albumList.size() == 0) {
            return null;
        }

        List<VideoIndexBean> result = new ArrayList<VideoIndexBean>();
        boolean ifBreak = false;
        for(int i = 0; i < maxLength; i++) {
            if(ifBreak) {
                break;
            }
            for(int j = 0; j < albumList.size(); j++) {
                if(albumList.get(j).size() > i) {
                    result.add(albumList.get(j).get(i));
                    if(result.size() >= CpInfoDataProcessTask.MAX_VIDEO) {
                        ifBreak = true;
                        break;
                    }
                }
            }
        }
        if(null == result || result.size() == 0) {
            return null;
        }
        return result;
    }

    /**
     * transform to thrift object from cp info pojo
     * @param cpInfoBean CpInfoBean cp info pojo
     * @return ContentProviderAttribute cp info thrift object.return null if have errors
     */
    private ContentProviderAttribute transformToThriftObject(CpInfoBean cpInfoBean) {
        if(null == cpInfoBean) {
            return null;
        }
        ContentProviderAttribute result = null;
        try {
            result = new ContentProviderAttribute();
            result.setLetv_original_id(cpInfoBean.getLetv_original_id());
            result.setUrl(cpInfoBean.getUrl());
            result.setSource(cpInfoBean.getSource());
            result.setName(cpInfoBean.getName());
            result.setDesc(cpInfoBean.getDesc());
            result.setLocation(cpInfoBean.getLocation());

            if(null != cpInfoBean.getThumbnail_list() && cpInfoBean.getThumbnail_list().size() > 0) {
                List<Thumbnail> thumbnail_list = new ArrayList<Thumbnail>();
                Thumbnail thumbnail = new Thumbnail();
                thumbnail.setUrl(cpInfoBean.getThumbnail_list().get(0).getUrl());
                thumbnail_list.add(thumbnail);
                result.setThumbnail_list(thumbnail_list);
            }

            result.setJoin_time(null == cpInfoBean.getJoin_time() ? 0l : cpInfoBean.getJoin_time().longValue());
            result.setComment_count(null == cpInfoBean.getComment_count() ? 0 : cpInfoBean.getComment_count().intValue());
            result.setPlay_count(null == cpInfoBean.getPlay_count() ? 0l : cpInfoBean.getPlay_count().longValue());
            result.setFans_count(null == cpInfoBean.getFans_count() ? 0l : cpInfoBean.getFans_count().longValue());
            result.setVideo_count(null == cpInfoBean.getVideo_count() ? 0 : cpInfoBean.getVideo_count().intValue());
            result.setAlbum_id_list(cpInfoBean.getAlbum_id_list());

            if(null != cpInfoBean.getVideo_index_list() && cpInfoBean.getVideo_index_list().size() > 0) {
                List<VideoIndex> video_index_list = new ArrayList<VideoIndex>();
                for(int i = 0; i < cpInfoBean.getVideo_index_list().size(); i++) {
                    VideoIndexBean videoIndexBean = cpInfoBean.getVideo_index_list().get(i);
                    VideoIndex videoIndex = new VideoIndex();
                    videoIndex.setAlbum_id(videoIndexBean.getAlbum_id());
                    videoIndex.setVideo_index(videoIndexBean.getVideo_index());
                    video_index_list.add(videoIndex);
                }
                result.setVideo_index_list(video_index_list);
            }
        } catch(Exception e) {
            LOGGER.error("transform to thrift object from cp info pojo have errors,cp_id=" + cpInfoBean.getId(), e);
        }
        return result;
    }

    /**
     * serialize thrift object
     * @param cPInfo ContentProviderAttribute, thrift object
     * @return serialize result.return null if have errors
     */
    private byte[] thriftSerialize(ContentProviderAttribute cPInfo) {
        if(null == cPInfo) {
            return null;
        }
        byte[] result = null;
        try {
            TMemoryBuffer tMemoryBuffer = new TMemoryBuffer(32);
            TBinaryProtocol tBinaryProtocol = new TBinaryProtocol(tMemoryBuffer);
            cPInfo.write(tBinaryProtocol);
            result = tMemoryBuffer.getArray();
        } catch(Exception e) {
            LOGGER.error("serialize thrift object have errors,letv_original_id=" + cPInfo.getLetv_original_id(), e);
            result = null;
        }
        return result;
    }
}
